package com.xujian.cdc.sink.starrocks;

import com.xujian.common.util.EnviromentUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author star xu
 * @date 2023/1/29 23:20
 * @Description:
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class Kafka2StarrockCalcuOrderWithSql {
    public static void main(String[] args) throws Exception {
        // 1 定义流式计算的环境
        String storgeType = "";
        final StreamExecutionEnvironment env = EnviromentUtil.build(storgeType);
        env.setParallelism(1);

        // 2 kafka source
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String strSourceSql = "\n" +
                "CREATE TABLE channel_order_source (\n" +
                "    id    bigint,\n" +
                "    `organization_name` STRING,\n" +
                "   `package_price` DOUBLE,\n" +
                "   `order_status` int,\n" +
                "   `order_no` STRING\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'whe_member_meinian_channel_order1',\n" +
                " 'properties.bootstrap.servers' = '192.168.16.77:9092',\n" +
                " 'properties.group.id' = 'my-test-group',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'properties.auto.offset.reset' = 'earliest',\n" +
                " 'format' = 'json'\n" +
                ")";
        String strSinkSql = "CREATE TABLE channel_order_calcu(\n" +
                "    organization_name STRING,\n" +
                "    order_count DOUBLE,\n" +
                "    order_amount DOUBLE,\n" +
                "    PRIMARY KEY (organization_name) NOT ENFORCED \n" +
                "  ) WITH (\n" +
                "    'connector' = 'starrocks',\n" +
                "    'jdbc-url' = 'jdbc:mysql://192.168.16.77:9030?characterEncoding=utf-8&useSSL=false',\n" +
                "    'load-url' = '192.168.16.77:8030',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '',\n" +
                "    'database-name' = 'starrocks',\n" +
                "    'table-name' = 'channel_order_calcu',\n" +
                "'sink.buffer-flush.interval-ms'='10000',\n" +
                "'sink.properties.format'='json',\n" +
                "'sink.properties.strip_outer_array'='true',\n" +
                "'sink.parallelism'='1'\n" +
                "  );";

        tableEnv.executeSql(strSourceSql);
        tableEnv.executeSql(strSinkSql);
        tableEnv.executeSql("SELECT organization_name,COUNT(*) order_count,SUM(package_price) order_amount \n" +
                "FROM (\n" +
                "SELECT\n" +
                "   order_no,\n" +
                "   organization_name,\n" +
                "   order_status,\n" +
                "   package_price,\n" +
                "   ROW_NUMBER() OVER(PARTITION BY order_no ORDER BY order_no DESC) AS rn \n" +
                "FROM channel_order_source\n" +
                ") t1 \n" +
                "where t1.rn=1\n" +
                " and order_status =2\n" +
                " GROUP BY t1.organization_name");
    }
}
